defmodule SockWorkstation.SockTTSClient do
  use WebSockex

  require Logger
  @url "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"
  # @output_file "output.pcm"
  @api_key "sk-16df948259d347f7a31c67e314c1a8fc"

  alias SockWorkstation.{
    TTSHelperGen,
    FrameHelperGen
  }

  def start_link({session_id, tts_text}) do
    {:ok, _pid} =
      WebSockex.start_link(
        @url,
        __MODULE__,
        %{session_id: session_id, tts_text: tts_text},
        extra_headers: headers()
        # debug: [:trace]
      )
  end

  defp headers do
    [
      {"Authorization", "bearer #{@api_key}"},
      {"X-DashScope-DataInspection", "enable"}
    ]
  end

  @impl WebSockex
  def handle_connect(_conn, state) do
    IO.puts("TTS WebSocket Connected!")
    # IO.inspect(state)
    self_pid = self()
    Process.send_after(self_pid, :run_tts_task, 120)

    state =
      state
      |> Map.put(:self_pid, self_pid)
      |> Map.put(:tts_task_id, "")
      |> Map.put(:bin_buffer, <<>>)
      |> Map.put(:buffer_size, 0)
      |> Map.put(:chunk_num, 0)

    {:ok, state}
  end

  @impl WebSockex
  def handle_frame({:text, message}, state) do
    message = Jason.decode!(message)
    # IO.inspect(state)
    case message["header"]["event"] do
      "task-started" ->
        IO.puts("tts task started")
        TTSHelperGen.tts_started(state.session_id)
        {:ok, Map.put(state, :tts_started, true)}

      "task-finished" ->
        IO.puts("tts task finished")

        send(state.self_pid, :tts_stream_end)
        # send(SockWorkstation.CoopMgrGen.get_state(state.session_id).coop_mgr_pid, :tts_result)
        # send(CoopMgrGen.get_state(state.session_id).coop_mgr_pid, :run_sent_asr)
        {:ok, state}

      "task-failed" ->
        IO.puts("tts task failed: #{message["header"]["error_message"]}")
        {:ok, state}

      _ ->
        {:ok, state}
    end
  end

  def handle_frame({:binary, pcm_data}, state) do
    # IO.puts("TTSClient Received pcm frame length: #{inspect(byte_size(pcm_data))}")
    # {:ok, opus_data} = SockWorkstation.OpusNif.encode_pcm_data(data)
    # {:ok, opus_data}=SockWorkstation.CodecNifGen.encode_pcm(state.session_id, data)
    # SockWorkstation.SockCoopClient.send_binary(state.sent_pid, opus_data)
    FrameHelperGen.fetch_pcmbin(state.session_id, pcm_data)
    # SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
    #   {:tts_pcm, state.session_id, pcm_data}
    # )

    {:ok,
     %{
       state
       | chunk_num: state.chunk_num + 1,
         buffer_size: state.buffer_size + byte_size(pcm_data)
     }}
  end

  @impl true

  def handle_info(:run_tts_task, state) do
    tts_task_id = UUID.uuid4(:hex)

    run_task_message = %{
      header: %{
        action: "run-task",
        task_id: tts_task_id,
        streaming: "duplex"
      },
      payload: %{
        task_group: "audio",
        task: "tts",
        function: "SpeechSynthesizer",
        model: "cosyvoice-v1",
        parameters: %{
          text_type: "PlainText",
          voice: "longxiaochun",
          format: "pcm",
          sample_rate: 16000,
          volume: 50,
          rate: 1,
          pitch: 1
        },
        input: %{}
      }
    }

    Process.send_after(state.self_pid, {:send_tts_text, state.tts_text}, 200)

    {:reply, {:text, Jason.encode!(run_task_message)},
     state |> Map.put(:tts_task_id, tts_task_id)}
  end

  def handle_info({:send_tts_text, text}, state) do
    continue_task_message = %{
      header: %{
        action: "continue-task",
        task_id: state.tts_task_id,
        streaming: "duplex"
      },
      payload: %{
        input: %{
          text: text
        }
      }
    }

    Process.send_after(state.self_pid, :finish_tts_task, 200)

    {:reply, {:text, Jason.encode!(continue_task_message)}, state}
  end

  def handle_info(:finish_tts_task, state) do
    finish_task_message = %{
      header: %{
        action: "finish-task",
        task_id: state.tts_task_id,
        streaming: "duplex"
      },
      payload: %{
        input: %{}
      }
    }

    {:reply, {:text, Jason.encode!(finish_task_message)}, state}
  end

  def handle_info(:tts_stream_end, state) do
    # IO.puts("Audio stream has ended")
    # TTSHelperGen.tts_stopped(state.session_id)

    time = calculate_duration(state.buffer_size)
    # time = due + 1200
    Process.sleep(time)
    SockWorkstation.CoopMgrGen.run_tts_result(state.session_id)

    Process.sleep(1200)
    send(state.self_pid, :task_finished)

    {:ok, %{state | bin_buffer: <<>>, tts_started: false}}
  end

  def handle_info(:task_finished, state) do
    IO.puts("TTS service stop")
    TTSHelperGen.tts_stopped(state.session_id)
    # SockWorkstation.CoopMgrGen.run_tts_result(state.session_id)
    {:close, state}
  end

  @impl WebSockex
  def handle_disconnect(_conn, state) do
    IO.puts("已断开与WebSocket服务器的连接")
    {:ok, state}
  end

  @impl WebSockex
  def terminate(reason, _state) do
    # CoopMgrGen.rm_tts_pid(state["session_id"])
    IO.puts("TTS WebSocket客户端终止，原因：#{inspect(reason)}")
    exit(:normal)
  end

  defp calculate_duration(buffer_size) do
    channels = 1
    sample_rate = 16000
    bit_depth = 16
    # 计算每个样本的字节数
    bytes_per_sample = div(bit_depth, 8)

    # 计算总样本数
    total_samples = div(buffer_size, bytes_per_sample)

    # 计算播放时间（秒）
    duration = div(total_samples, sample_rate * channels)
    duration * 1000
  end
end

# def handle_info({:pcm_data, data}, state) do
#   #  IO.puts("Received pcm data length: #{inspect(byte_size(data))}")
#   #  SockWorkstation.PcmEncoderGen.pcm_to_opus(state.session_id, data)
#   # bin_buffer = [data | state.bin_buffer]
#   # send(state.sent_pid, {:tts_pcm, data})
#   bin_buffer = <<state.bin_buffer::binary, data::binary>>

#   if state.chunk_num >=6 do
#     send(state.self_pid, {:pcm_chunk, bin_buffer})
#     {:ok, state}
#   end

#   # buffer = :queue.in(data, state.buffer)
#   {:ok, %{state | bin_buffer: bin_buffer}}
#   # {:ok, state}
# end

# def handle_info({:pcm_chunk, chunk}, state) do
#   # IO.puts("Received pcm chunk length: #{inspect(byte_size(chunk))}")
#   SockWorkstation.PcmEncoderGen.pcm_to_opus(state.session_id, chunk)
#   # {:ok, opus_data}=SockWorkstation.CodecNifGen.encode_pcm(state.session_id, chunk)
#   # SockWorkstation.SockCoopClient.send_binary(state.sent_pid, opus_data)

#   # send(state.sent_pid, {:tts_pcm, chunk})
#   {:ok, %{state | bin_buffer: <<>>, chunk_num: 0}}
# end

# def handle_info(:tts_stream_end, state) do
#   # IO.puts("Audio stream has ended")
#   # pcm_data = :lists.reverse(state.bin_buffer)
#   # SockWorkstation.PcmEncoderGen.pcm_to_opus(state.session_id, state.bin_buffer)
#   # {:ok, opus_data}=SockWorkstation.CodecNifGen.encode_pcm(state.session_id,state.bin_buffer)
#   # SockWorkstation.SockCoopClient.send_binary(state.sent_pid, opus_data)
#   # send(state.sent_pid, {:tts_pcm, state.bin_buffer})
#   time = calculate_duration(state.buffer_size)
#   # time = due + 1200
#   # IO.puts("Audio stream duration: #{time} seconds")
#   # Process.send_after(state.self_pid, :asr_start, due)

#   # Process.send_after(
#   #   SockWorkstation.CoopMgrGen.get_state(state.session_id).coop_mgr_pid,
#   #   :tts_result,
#   #   due
#   # )
#   Process.sleep(time)
#   SockWorkstation.CoopMgrGen.run_tts_result(state.session_id)
#   Process.sleep(1200)
#   send(state.self_pid, :task_finished)

#   {:ok, %{state | bin_buffer: <<>>, tts_started: false}}
# end

# def send_run_task(pid, task_id) do
#   # Process.sleep(20)

#   message = %{
#     header: %{
#       action: "run-task",
#       task_id: task_id,
#       streaming: "duplex"
#     },
#     payload: %{
#       task_group: "audio",
#       task: "tts",
#       function: "SpeechSynthesizer",
#       model: "cosyvoice-v1",
#       parameters: %{
#         text_type: "PlainText",
#         voice: "longxiaochun",
#         format: "pcm",
#         sample_rate: 16000,
#         volume: 50,
#         rate: 1,
#         pitch: 1
#       },
#       input: %{}
#     }
#   }

#   WebSockex.send_frame(pid, {:text, message |> Jason.encode!()})
# end

# def send_tts_task(pid, task_id, text) do
# IO.puts("send_tts_task#{IO.inspect(texts)}")
# Process.sleep(500)

# continue_task_message = %{
#   header: %{
#     action: "continue-task",
#     task_id: task_id,
#     streaming: "duplex"
#   },
#   payload: %{
#     input: %{
#       text: text
#     }
#   }
# }

#   WebSockex.send_frame(
#     pid,
#     {:text, Jason.encode!(continue_task_message)}
#   )
# end

# def send_finish_task(pid, task_id) do
# Process.sleep(600)

#   finish_task_message = %{
#     header: %{
#       action: "finish-task",
#       task_id: task_id,
#       streaming: "duplex"
#     },
#     payload: %{
#       input: %{}
#     }
#   }

#   WebSockex.send_frame(
#     pid,
#     {:text, Jason.encode!(finish_task_message)}
#   )
# end

# defp split_by_punctuation(text) do
#   # 正则表达式匹配中文标点符号
#   regex = ~r/[。！？，]+/u

#   # 使用 String.split/3 分割字符串
#   String.split(text, regex, trim: true)
# end
# def handle_info(:close, state) do
#   # 在连接关闭时发送最后一小段缓存的音频数据
#   IO.puts("正在发送最后一小段缓存的音频数据...")
#   SockWorkstation.OpusCodec.send_pcm_data(state.resv_pid, state.buffer)
#   {:stop, :normal, state}
# end

# def run_task(text) do
#   SockWorkstation.SockTTSClient.start_link()
#   Process.sleep(100)
#   send_run_task()
#   # Process.sleep(500)
#   #  text = "床前明月光,疑是地上霜,举头望明月,低头思故乡"
#   send_tts_task(text)
#   # Process.sleep(500)
#   send_finish_task()
# end

# @impl true
# def handle_info(:run_task, state) do
#   # send_run_task(state.pid, state.task_id)

#   Logger.info("run_task#{IO.inspect(state)}")
#   # frame = %{
#   #   header: %{
#   #     action: "run-task",
#   #     task_id: state.task_id,
#   #     streaming: "duplex"
#   #   },
#   #   payload: %{
#   #     task_group: "audio",
#   #     task: "tts",
#   #     function: "SpeechSynthesizer",
#   #     model: "cosyvoice-v1",
#   #     parameters: %{
#   #       text_type: "PlainText",
#   #       voice: "longxiaochun",
#   #       format: "pcm",
#   #       sample_rate: 16000,
#   #       volume: 50,
#   #       rate: 1,
#   #       pitch: 1
#   #     },
#   #     input: %{}
#   #   }
#   # }

#   # Process.send_after(state.pid, :tts_started, 20)
#   {:noreply, state}
#   # {:reply, frame, state}
# end

# def handle_info(:tts_started, state) do
#   # send_tts_task(state.pid, state.task_id, state.message)
#   Logger.info("start tts#{IO.inspect(state)}")
#   # IO.inspect(state)
#   # frame = %{
#   #   header: %{
#   #     action: "continue-task",
#   #     task_id: state.task_id,
#   #     streaming: "duplex"
#   #   },
#   #   payload: %{
#   #     input: %{
#   #       text: state.message
#   #     }
#   #   }
#   # }

#   # Process.send_after(state.pid, :finish_task, 5000)

#   # {:reply, frame, state}
#   {:noreply, state}
# end

# def handle_info(:finish_task, state) do
#   Logger.info("finish_task#{IO.inspect(state)}")
#   # frame = %{
#   #   header: %{
#   #     action: "finish-task",
#   #     task_id: state.task_id,
#   #     streaming: "duplex"
#   #   },
#   #   payload: %{
#   #     input: %{}
#   #   }
#   # }
#   {:noreply, state}
#   # {:reply, frame, state}
# end
